package com.naiterui.ehp.bs.im.service;

import cn.hutool.system.HostInfo;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import com.naiterui.ehp.bs.im.config.ListenerConfig;
import com.naiterui.ehp.bs.im.config.MqttConnectionConfig;
import com.naiterui.ehp.bs.im.network.mqtt.async.ClientStatusCallback;
import com.naiterui.ehp.bs.im.network.mqtt.handler.ClientStatusHandler;
import com.naiterui.ehp.bs.im.network.mqtt.pool.MqttConnection;

/**
 * 客户端状态监听类
 */
@Service
public class StartupListener implements InitializingBean {

    /**
     * logger
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(StartupListener.class);


    /**
     * 客户端状态操作类
     */
    private final ClientStatusHandler clientStatusHandler;


    private final MqttConnectionConfig mqttConnectionConfig;

    /**
     * 监听配置
     */
    private final ListenerConfig listenerConfig;

    /**
     * 上下线连接Map：<code>monitorConnection</code>
     */
    private Map<String, MqttConnection> monitorConnMap = new HashMap<>();

    public StartupListener(ClientStatusHandler clientStatusHandler, MqttConnectionConfig mqttConnectionConfig, ListenerConfig listenerConfig) {
        this.clientStatusHandler = clientStatusHandler;
        this.mqttConnectionConfig = mqttConnectionConfig;
        this.listenerConfig = listenerConfig;
    }


    /**
     * spring容器启动时,初始化一些信息
     */
    @Override
    public void afterPropertiesSet() {
        /** 初始化mqtt连接池配置 **/
        this.initClientMonitor();
    }

    /**
     * 客户端上下线状态监控
     */
    private void initClientMonitor() {
        try {
            /** 初始化client监控连接 **/
            Set<String> serverSet = new HashSet<>();
            Collections.addAll(serverSet, this.mqttConnectionConfig.getInnerServers());
            this.monitorClient(serverSet);

            // 启动连接状态监视线程
            ClientMonitor monitor = new ClientMonitor();
            monitor.start();

        } catch (Exception e) {
            LOGGER.error("initClientMoniter ERROR; ", e);
        }
    }

    /**
     * client上下线状态监控
     */
    private void validateClient() {
        if (null == this.monitorConnMap || this.monitorConnMap.isEmpty()) {
            return;
        }

        Set<String> errorServer = new HashSet<>();
        for (Entry<String, MqttConnection> tempEntry : this.monitorConnMap.entrySet()) {
            String mqttServer = tempEntry.getKey();
            MqttConnection mqttConn = tempEntry.getValue();
            if (null == mqttConn.getMqttClient() || mqttConn.getMqttClient().isConnected()) {
                continue;
            }
            errorServer.add(mqttServer);
            try {
                mqttConn.getMqttClient().close();
            } catch (Exception e) {
                LOGGER.error("validateClientMonitor ERROR ;", e);
            }
        }

        this.monitorClient(errorServer);
    }

    /**
     * 建立mqtt连接监听client上下线
     */
    private void monitorClient(Set<String> mqttServers) {
        HostInfo hostInfo = new HostInfo();
        for (String mqttServer : mqttServers) {
            try {

                // 生成唯一clientId
                String clientId = hostInfo.getAddress() + "_ClientMonitor";
                // 生成MQTT连接
                MqttClient mqttClient = new MqttClient(mqttServer, clientId);
                // 建立连接
                MqttConnectOptions connOpts = new MqttConnectOptions();
                connOpts.setUserName(this.mqttConnectionConfig.getUserName());
                connOpts.setPassword(this.mqttConnectionConfig.getPassword().toCharArray());
                connOpts.setCleanSession(this.mqttConnectionConfig.isCleanSession());
                connOpts.setConnectionTimeout(this.mqttConnectionConfig.getConnectionTimeout());
                connOpts.setKeepAliveInterval(this.mqttConnectionConfig.getKeepAliveInterval());
                mqttClient.connect(connOpts);
                // 构建mqttConnection对象
                MqttConnection mqttConnection = new MqttConnection(mqttClient);

                // 监控客户端上下线状态
                mqttConnection.subscribe(this.listenerConfig.getShareGroupTopicPrefix() + this.listenerConfig.getStatusLsnTopic(), 2);
                mqttConnection.subscribe(this.listenerConfig.getShareGroupTopicPrefix() + this.listenerConfig.getWillLsnTopic(), 2);
                mqttConnection.setCallBack(new ClientStatusCallback(this.clientStatusHandler));

                this.monitorConnMap.put(mqttServer, mqttConnection);

                LOGGER.info("{} start to listen mqtt {}", hostInfo.getAddress(), mqttServer);
            } catch (Exception e) {
                LOGGER.error("{} start to listen mqtt {} ERROR", hostInfo.getAddress(), mqttServer);
            }
        }

    }

    /**
     * 每隔3分钟检查client监控连接状态
     */
    public class ClientMonitor extends Thread {

        @Override
        public void run() {
            while (true) {
                try {
                    // 检查时间间隔: 3分钟
                    long checkInterval = 180000L;
                    sleep(checkInterval);
                } catch (InterruptedException e) {
                    LOGGER.error("ClientMonitor encountered, exit.");
                    return;
                }
                try {
                    StartupListener.this.validateClient();
                } catch (Exception e) {
                    LOGGER.error("failed to monitor ClientMonitor ." + e.getMessage() + ",continue to monitor.");
                }
            }

        }

    }
}
